package com.kj.misc.service;

import java.text.SimpleDateFormat;
import java.util.LinkedList;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.eventbus.Subscribe;
import com.ivan.toolkit.common.Module;
import com.ivan.toolkit.common.Modules;
import com.kj.misc.domain.MiscConfig;
import com.kj.misc.domain.MiscFusionConfig;
import com.kj.misc.domain.MiscStoreType;
import com.kj.misc.domain.MiscValue;

import lombok.Data;
import lombok.extern.log4j.Log4j2;

@Component
@Log4j2
public class MiscDataFusion extends Modules{
	
	public static final String TRANSERROR_CURRENT = "TRANSERROR.CURRENT";
	public static final String TRANSERROR_10MIN = "TRANSERROR.10MIN";
	public static final String TRANSERROR_30MIN = "TRANSERROR.30MIN";
	public static final String TRANSERROR_60MIN = "TRANSERROR.60MIN";
	
	public static final String HEADFLOW_15MIN = "HEADFLOW.15MIN";
	
	public static final String TAILFLOW_15MIN = "TAILFLOW.15MIN";
	
	public static final long LiveMilliseconds = 7260000;
	
	private static SimpleDateFormat timeFmt = 
			new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:sss");
	
	
	@Autowired
	private MiscData data;
	
	private StatisticsSession headSession = new StatisticsSession();
	private StatisticsSession tailSession = new StatisticsSession();
	
	public MiscDataFusion() {
		
		this.register(new TransmissionError());
		this.register(new StatisticsModule());
		
	}

	
	@Override
	public void initialize() {
		
		data.registerData(TRANSERROR_CURRENT, MiscStoreType.period);
		data.registerData(TRANSERROR_10MIN, MiscStoreType.period);
		data.registerData(TRANSERROR_30MIN, MiscStoreType.period);
		data.registerData(TRANSERROR_60MIN, MiscStoreType.period);
		
		data.registerData(HEADFLOW_15MIN, MiscStoreType.update);
		data.registerData(TAILFLOW_15MIN, MiscStoreType.update);
		
		super.initialize();
		
	}
	

	private MiscFusionConfig getConfig() {
		return MiscConfig.instance.get().getFusionConfig();
	}
	
	// 累积流量积算
	@Scheduled(cron="0 0/15 *  * * ? ")
	public void schedule15min() {
		
		long now = System.currentTimeMillis() / 1000 * 1000;
		
		long begin = now - 900000;
		long end = now -1;
		
		MiscValue vHead = MiscValue.at(begin, headSession.calculate(begin, end));
		data.update(HEADFLOW_15MIN, vHead);
		log.info(
				"Head15=>begin:" + timeFmt.format(begin) + 
				",end:" + timeFmt.format(end) + 
				",value:" + vHead.getValue());
		
		MiscValue vTail = MiscValue.at(begin, tailSession.calculate(begin, end));
		data.update(TAILFLOW_15MIN, vTail);
		log.info(
				"Tail15=>begin:" + timeFmt.format(begin) + 
				",end:" + timeFmt.format(end) + 
				",value:" + vTail.getValue());

	}
	
	// 累积流量积算
	@Scheduled(cron="0/10 * *  * * ? ")
	public void scheduleTransmissionError() {
		
		long now = System.currentTimeMillis() / 1000 * 1000;
		
		data.update(TRANSERROR_10MIN, MiscValue.at(now, calculateTransmissionError(now - 600000, now)));
		data.update(TRANSERROR_30MIN, MiscValue.at(now, calculateTransmissionError(now - 1800000, now)));
		data.update(TRANSERROR_60MIN, MiscValue.at(now, calculateTransmissionError(now - 3600000, now)));

	}
	
	private double calculateTransmissionError(long begin, long end) {
		
		double headFs = headSession.calculateWithPeriodCheck(begin, end);
		double tailFs = tailSession.calculateWithPeriodCheck(begin, end);
		return tailFs - headFs;
		
	}
	
	public class TransmissionError implements Module {
		
		private MiscValue currentHead = null;
		private MiscValue currentTail = null;
	
		
		@Override
		public void initialize() {
			data.getEventBus().register(this);
		}

		@Override
		public void terminate() {
			data.getEventBus().unregister(this);
		}
		
		@Subscribe
		public void dispatchMiscEvent(MiscDataEvent event) {
			
			String key = event.getKey();
			
			if(Strings.isNullOrEmpty(key))
				return;
			
			boolean updated = false;
			
			if(key.equals(getConfig().getHeadKey())) {
				
				if(event.getNewValue() != null) {
					currentHead = event.getNewValue();
					updated = true;
				}
				
			} 
			
			if(key.equals(getConfig().getTailKey())) {
				
				if(event.getNewValue() != null) {
					currentTail = event.getNewValue();
					updated = true;
				}
				
			} 
			
			if(updated) valueUpdate();
			
		}
		
		private void valueUpdate() {
			
			if(currentHead == null || Double.isNaN(currentHead.getValue())) {
				log.info("Head value is invalid.");
				data.update(TRANSERROR_CURRENT, Double.NaN);
				return;
			}
			
			if(currentTail == null || Double.isNaN(currentTail.getValue())) {
				log.info("Tail value is invalid.");
				data.update(TRANSERROR_CURRENT, Double.NaN);
				return;
			}
			
			long timeDelta = Math.abs(currentTail.getBirth() - currentHead.getBirth());
			
			if(timeDelta > getConfig().getCurrentLimitMS()) {
				log.info("Birth delta is out of limit.");
				data.update(TRANSERROR_CURRENT, Double.NaN);
				return;
			}
			
			// 计算当前输差
			data.update(TRANSERROR_CURRENT, currentTail.getValue() - currentHead.getValue());
			
		}

	}
	

	@Data
	class StatisticsValue {
		
		private void calculateTotal() {
			total = (end - begin) * speed / 3600000;
		}
		
		public double calculateTotal(long delta) {
			return delta * speed / 3600000;
		}
		
		public long getPeriod() {
			return end - begin;
		}
		
		private double speed;
		
		private long begin;
		
		private long end;
		
		private double total;
		
	}

	
	class StatisticsLine {
		
		private LinkedList<StatisticsValue> values = Lists.newLinkedList();
		
		public StatisticsValue getLast() {
			return values.peekLast();
		}
		
		public StatisticsValue getFirst() {
			return values.peekFirst();
		}
		
		private StatisticsValue buildStatisticsValue(MiscValue mv, StatisticsValue prev) {
			
			if(prev == null) {
				
				StatisticsValue sv = new StatisticsValue();
				sv.setSpeed(mv.getValue());
				sv.setBegin(mv.getBirth() - LiveMilliseconds);
				sv.setEnd(mv.getBirth());
				sv.calculateTotal();
				return sv;
				
			} else {
				
				StatisticsValue sv = new StatisticsValue();
				sv.setSpeed(mv.getValue());
				sv.setBegin(prev.getEnd());
				sv.setEnd(mv.getBirth());
				sv.calculateTotal();
				return sv;
				
			}
						
		}
		
		public void feed(MiscValue value) {
			
			if(value == null) return;
			
			StatisticsValue last = getLast();
			
			StatisticsValue sv = buildStatisticsValue(value, last);

			values.add(sv);
		}
		
		public void filterBeforeThan(long milliseconds) {
			
			values.removeIf(v->v.getEnd() < milliseconds);
			
		}
		
		public double calculateFlowWithPeriodCheck(long begin, long end) {
			
			return calculateFlow(begin, end, true);
			
			/*
			LinkedList<StatisticsValue> includes = Lists.newLinkedList();
			for(StatisticsValue v : values) {
				if(isIncluded(begin, end, v)) {
					includes.add(v);							
				}
			}
			
			// 如果没有符合条件的点，返回NAN
			if(includes.isEmpty()) return Double.NaN;
			
			// 判断第一个和最后一个值是否有效
			StatisticsValue first = includes.getFirst();
			if((first.getEnd() - begin) > getConfig().getIntegratLimitMS()) return Double.NaN;
			StatisticsValue last = includes.getLast();
			if((end - last.getEnd()) > getConfig().getIntegratLimitMS()) return Double.NaN;
			
			double sum = last.calculateTotal(end - last.getEnd());
			for(StatisticsValue v : includes) {
				
				if(v.getPeriod() > getConfig().getIntegratLimitMS())
					return Double.NaN;
				
				sum += calculateFlow(begin, end, v);
				
			}

			return sum;
			*/
		}

		public double calculateFlow(long begin, long end) {
			
			return calculateFlow(begin, end, false);
			
			/*
			// 保存最接近end的值
			StatisticsValue max = null;
			
			double sum = 0.0;
			
			boolean valid = false;
			
			for(StatisticsValue v : values) {
				
				if(isIncluded(begin, end, v)) {
					
					sum += calculateFlow(begin, end, v);
					
					if(max == null) {
						max = v;
					} else {
						if(v.getEnd() > max.getEnd())
							max = v;
					}
					
					valid = true;
					
				}
				
			}
			
			// 如果没有符合条件的点，返回NAN
			if(!valid) return Double.NaN;
			
			// 推算最后一段的量
			if(max != null) {
				sum += max.calculateTotal(end - max.getEnd());
			}
			
			return sum;
			*/
		}
		
		public double calculateFlow(long begin, long end, boolean periodCheck) {
			
			LinkedList<StatisticsValue> includes = Lists.newLinkedList();
			for(StatisticsValue v : values) {
				if(isIncluded(begin, end, v)) {
					includes.add(v);							
				}
			}
			
			// 如果没有符合条件的点，返回NAN
			if(includes.isEmpty()) return Double.NaN;
			
			
			StatisticsValue first = includes.getFirst();
			StatisticsValue last = includes.getLast();
			
			if(periodCheck) {
				// 判断第一个和最后一个值是否有效
				if((first.getEnd() - begin) > getConfig().getIntegratLimitMS()) return Double.NaN;
				if((end - last.getEnd()) > getConfig().getIntegratLimitMS()) return Double.NaN;
			}

			
			double sum = last.calculateTotal(end - last.getEnd());
			for(StatisticsValue v : includes) {
				
				if(periodCheck) {
					if(v.getPeriod() > getConfig().getIntegratLimitMS())
						return Double.NaN;
				}
				
				sum += calculateFlow(begin, end, v);
				
			}

			return sum;
			
		}
		
		private boolean isIncluded(long begin, long end, StatisticsValue value) {
			if(value.getEnd() < begin) return false;
			if(value.getEnd() > end) return false;
			if(value.getEnd() == value.getBegin()) return false;
			return true;
		}
		
		private double calculateFlow(long begin, long end, StatisticsValue value) {
			
			long start = Math.max(begin, value.getBegin());
			long stop = Math.min(end, value.getEnd());
			
			return value.getTotal() * (stop - start) / (value.getEnd() - value.getBegin());
			
		}
		
	}
	
	class StatisticsSession {
		
		private StatisticsLine line = new StatisticsLine();
		
		public void feed(MiscValue mv) {
			
			if(mv != null) {
				
				line.feed(mv);
				
				// 过滤2小时以前的数据(多留了60秒，以防万一)
				line.filterBeforeThan(mv.getBirth() - LiveMilliseconds);
				
			}

		}
		
		public double calculateWithPeriodCheck(long begin, long end) {
			try {
				return line.calculateFlowWithPeriodCheck(begin , end);
			} catch(Exception e) {
				return Double.NaN;
			}
		}
		
		public double calculate(long begin, long end) {

			return line.calculateFlow(begin, end);
			
		}
		
	}

	
	class StatisticsModule implements Module {

		public StatisticsModule() {
		}
		
		@Override
		public void initialize() {
			data.getEventBus().register(this);
		}

		@Override
		public void terminate() {
			data.getEventBus().unregister(this);
		}
		
		@Subscribe
		public void dispatchMiscEvent(MiscDataEvent event) {
			
			String key = event.getKey();
			
			if(Strings.isNullOrEmpty(key))
				return;
			
			if(key.equals(getConfig().getHeadKey())) {
				headSession.feed(event.getNewValue());
			} 
			
			if(key.equals(getConfig().getTailKey())) {
				tailSession.feed(event.getNewValue());
			}
			
		}
		
	}
	
	

}
